package service

import (
	"context"
	"github.com/go-kratos/kratos/v2/log"
	"github.com/gorilla/websocket"
	"go_private_im/api/protocol"
	"go_private_im/internal/comet/biz"
	"go_private_im/pkg/strings"
	"io"
	"math/rand"
	"net/http"
	"go_private_im/internal/comet/conf"
	"time"
)

var (
	timer    *time.Timer
	upgrader = websocket.Upgrader{
		ReadBufferSize:  1024,
		WriteBufferSize: 1024,
		CheckOrigin: func(r *http.Request) bool {
			return true
		},
	}
)

const (
	minServerHeartbeat = time.Minute * 10
	maxServerHeartbeat = time.Minute * 30
)

type WsHandlerService struct {
	uc   *biz.CometUsecase
	log  *log.Helper
	conf *conf.Config
}

func NewWsHandler(logger log.Logger, conf *conf.Config, uc *biz.CometUsecase) *WsHandlerService {
	return &WsHandlerService{log: log.NewHelper(logger), conf: conf, uc: uc}
}
func (w *WsHandlerService) Handler(wr http.ResponseWriter, r *http.Request) {
	var (
		ws       *websocket.Conn
		err      error
		authInfo *biz.AuthInfo
		ctx      = context.Background()
	)

	ws, err = upgrader.Upgrade(wr, r, nil)
	if err != nil {
		w.log.Info("upgrade:", err)
		return
	}
	defer ws.Close()

	timer = time.AfterFunc(w.conf.Protocol.HandshakeTimeout, func() {
		ws.Close()
	})
	defer timer.Stop()
	//http 升级 到 websocket
	//首次建立链接
	proto := &protocol.Proto{}

	if err := proto.ReadWebsocket(ws); err != nil {
		w.OpProtoFinish(ws, proto, err)
		return
	}
	if !proto.IsOpAuth() {
		w.OpProtoFinish(ws, proto, err)
		return
	}
	//效验权限
	if authInfo, err = w.WsAuth(ctx, ws ,proto); err != nil {
		w.OpProtoFinish(ws, proto, err)
		return
	}
	go w.dispatchWebsocket(ws, authInfo) //处理信息
	//监听ws 信息
	if err := w.AcceptWebsocket(ctx,ws,timer,authInfo); err != nil {
		authInfo.Ch.Close()
		authInfo.Bucket.Del(authInfo.Ch)
		ws.Close()
	}
}

func (w *WsHandlerService) AcceptWebsocket (ctx context.Context,ws *websocket.Conn,timer *time.Timer,authInfo *biz.AuthInfo) (err error) {
	lastHB := time.Now()
	proto := &protocol.Proto{}
	for {
		//从环中获取一个
		if proto, err = authInfo.Ch.CliProto.Set(); err != nil {
			break
		}
		if err = proto.ReadWebsocket(ws); err != nil {
			w.log.Errorf("read error", err)
			break
		}
		if proto.IsOpHeartbeat() {
			timer.Reset(authInfo.Hb)
			lastHB = w.ResetHeartbeat(ctx,authInfo,proto,lastHB)
		} else {
			//其他消息的接受
			if err = w.WsOperate(ctx, authInfo, proto); err != nil {
				break
			}
		}

		w.log.Infof("key: %s process proto:%v\n", authInfo.Key, proto)
		//使用完成归还
		authInfo.Ch.CliProto.SetAdv()
		authInfo.Ch.Signal()
	}
	return
}

func (w WsHandlerService) ResetHeartbeat(ctx context.Context,authInfo *biz.AuthInfo,proto *protocol.Proto,lastHB time.Time) time.Time  {
	proto.SetOpHeartbeatReply()
	if now := time.Now(); now.Sub(lastHB) > w.RandServerHearbeat() {
		if err1 := w.WsHeartbeat(ctx, authInfo); err1 == nil {
			lastHB = now
		}
		w.log.Infof("websocket heartbeat receive key:%s, UserId:%d", authInfo.Key, authInfo.UserId)
	}
	return lastHB
}

func (w *WsHandlerService) WsAuth(ctx context.Context,ws *websocket.Conn, proto *protocol.Proto) (*biz.AuthInfo, error) {
	authInfo, err := w.uc.Auth(ctx, proto)
	if err != nil {
		return nil, err
	}
	proto.SetOpAuthReply()
	if err = proto.WriteWebsocket(ws); err != nil {
		w.log.Errorf("key: %s mid : %d close", authInfo.Key, authInfo.UserId)
	}
	return authInfo, nil

}

func (w *WsHandlerService) OpProtoFinish(ws *websocket.Conn, proto *protocol.Proto, err error) {
	w.log.Errorf("start read error", err)
	proto.Op = protocol.OpProtoFinish
	proto.Body = nil
	proto.WriteWebsocket(ws)
	ws.Close()
}

func (s *WsHandlerService) RandServerHearbeat() time.Duration {
	return (minServerHeartbeat + time.Duration(rand.Int63n(int64(maxServerHeartbeat-minServerHeartbeat))))
}

func (w *WsHandlerService) dispatchWebsocket(ws *websocket.Conn, authInfo *biz.AuthInfo) {
	var (
		err    error
		finish bool
		online int32
	)
	w.log.Infof("key: %s start dispatch tcp goroutine", authInfo.Key)

	for {
		p := authInfo.Ch.Ready()
		w.log.Infof("key: %s proto ready", authInfo.Key,p)

		switch p {

		case protocol.ProtoFinish:
			w.log.Infof("key: %s receive proto finish\n", authInfo.Key)
			finish = true

			goto failed
		case protocol.ProtoReady:
			for {
				if p, err = authInfo.Ch.CliProto.Get(); err != nil {
					break
				}
				if p.Op == protocol.OpHeartbeatReply {
					if authInfo.Ch.Room != nil {
						online = authInfo.Ch.Room.OnlineNum()
					}
					//活的心跳 并处理
					if err = p.WriteWebsocketHeart(ws, online); err != nil {
						goto failed
					}
				} else {
					if err := p.WriteWebsocket(ws); err != nil {
						goto failed
					}
				}
				p.Body = nil // avoid memory leak

				authInfo.Ch.CliProto.GetAdv()
			}
		default:
			if err = p.WriteWebsocket(ws); err != nil {
				goto failed
			}
			w.log.Infof("websocket sent a message key:%s UserId:%d proto:%+v", authInfo.Key, authInfo.UserId, p)
		}
	}
failed:
	if err != nil && err != io.EOF {
		w.log.Errorf("key: %s dispatch  ws error(%v)", authInfo.Key, err)
	}
	ws.Close()
	for !finish {
		finish = (authInfo.Ch.Ready() == protocol.ProtoFinish)
	}
}

//处理其他请求
func (w *WsHandlerService) WsOperate(ctx context.Context, authInfo *biz.AuthInfo, proto *protocol.Proto) (err error) {
	switch proto.Op {
	case protocol.OpChangeRoom:
		if err := authInfo.Bucket.ChangeRoom(string(proto.Body), authInfo.Ch); err != nil {
			w.log.Errorf("b.ChangeRoom(%s) error(%v)", proto.Body, err)
		}
		proto.Op = protocol.OpChangeRoomReply
	case protocol.OpSub:
		if ops, err := strings.SplitInt32s(string(proto.Body), ","); err == nil {
			authInfo.Ch.Watch(ops...)
		}
		proto.Op = protocol.OpSubReply
	case protocol.OpUnsub:
		if ops, err := strings.SplitInt32s(string(proto.Body), ","); err == nil {
			authInfo.Ch.UnWatch(ops...)
		}
		proto.Op = protocol.OpUnsubReply
	default:
		// TODO ack ok&failed
		if err := w.uc.Receive(ctx, authInfo, proto); err != nil {
			w.log.Errorf("s.Report(%d) op:%d error(%v)", authInfo.UserId, proto.Op, err)
		}
		proto.Body = nil
	}
	return
}

func (w *WsHandlerService) WsHeartbeat(ctx context.Context, authInfo *biz.AuthInfo) (err error) {
	err = w.uc.Heartbeat(ctx, authInfo)
	return
}